package com.shujia.mllib

import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo04Image {
  def main(args: Array[String]): Unit = {
    // 手写数字识别
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[*]")
      .appName("Demo04Image")
      .getOrCreate()

    import spark.implicits._

    val imageDF: DataFrame = spark
      .read
      .format("image")
      .load("C:\\Users\\SHUJIA\\Desktop\\train")

    imageDF.show()
    imageDF.printSchema()

    val clearDF: DataFrame = imageDF
      .select($"image.origin" as "path", $"image.data" as "data")
      .rdd
      .map(row => {
        val fileName: String = row.getAs[String]("path").split("/").reverse.head
        val dataList: List[Byte] = row.getAs[Array[Byte]]("data").toList
        val newDataList: List[Double] = dataList.map(i => {
          if (i >= 0) {
            0.0
          } else {
            255.0
          }
        })
        (fileName, Vectors.dense(newDataList.toArray).toSparse)
      }).toDF("fileName", "features")

    // 加载image_res.txt数据 用于关联获取label
    val imageResDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("fileName String,label Double")
      .load("spark/data/mllib/data/image_res.txt")

    clearDF.join(imageResDF,"fileName").show()

    // 将做好数据特征工程的数据保存起来
    clearDF.join(imageResDF,"fileName")
      .select($"label",$"features")
      .write
      .format("libsvm")
      .mode(SaveMode.Overwrite)
      .save("spark/data/mllib/data/imagelibsvm")



  }

}
